#pragma once

#include <iostream>
#include <memory>
#include <mutex>
#include <cassert>
#include <list>
#include <unordered_map>
#include "../mqcommon/mq_helper.hpp"
#include "../mqcommon/mq_logger.hpp"
#include "../mqcommon/mq_msg.pb.h"

namespace bitmq
{
    #define DATAFILE_SUBFIX ".mqd"
    #define TEMPFILE_SUBFIX ".mqd.temp"

    using MessagePtr = std::shared_ptr<bitmq::Message>;
    class MessageMapper
    {
    public:
        MessageMapper()
        {}
        
        MessageMapper(std::string& basedir, const std::string& qname)
            : _qname(qname)
        {
            if (basedir.back() != '/')
                basedir += '/';
            _datafile = basedir + qname + DATAFILE_SUBFIX;
            _tempfile = basedir + qname + TEMPFILE_SUBFIX;
            if (FileHelper(basedir).exists() == false)
                assert(FileHelper::createDirectory(basedir));
            createMsgFile();
        }

        //1）创建队列消息文件
        bool createMsgFile()
        {
            //已经存在则不需要创建
            if (FileHelper(_datafile).exists() == true)
                return true;

            bool ret = FileHelper::createFile(_datafile);
            if (ret == false)
            {
                DLOG("创建队列数据文件 %s 失败", _datafile.c_str());
                return false;
            }
            return true;
        }

        //2）删除队列消息文件
        void removeMsgFile()
        {
            FileHelper::removeFile(_datafile);
            FileHelper::removeFile(_tempfile);
        }

        //3）队列消息的新增持久化
        bool insert(MessagePtr& msg)
        {
            return insert(_datafile, msg);
        }

        //4）队列消息的删除持久化
        bool remove(const MessagePtr& msg)
        {
            //1.将msg中的valid修改成'0'（注意这里的valid是string类型）
            msg->mutable_payload()->set_valid("0");
            //2.将msg进行序列化
            std::string body = msg->payload().SerializeAsString();
            if (body.size() != msg->length())
            {
                DLOG("不能修改文件中的数据信息，因为新生成的数据与原数据长度不一致");
                return false;
            }
            //3.将序列化的消息，写入到数据在文件中的指定位置
            FileHelper helper(_datafile);
            bool ret = helper.write(body.c_str(), msg->offset(), body.size());
            if (ret == false)
            {
                DLOG("向队列数据文件写入数据失败！");
                return false;
            }
            return true;
        }

        //5）提供持久化内容的垃圾回收
        std::list<MessagePtr> gc()
        {
            //1.加载出文件中所有的有效数据
            std::list<MessagePtr> result = load();   
            //DLOG("垃圾回收得到的有效消息数量：%d", result.size());
            //2.将有效数据，进行序列化存储到临时文件中
            FileHelper::createFile(_tempfile);
            for (auto& msg : result)
            {
                if (insert(_tempfile, msg) == false)
                {
                    DLOG("向临时文件写入消息数据失败!");
                    return result;
                }
            }
            //DLOG("垃圾回收后，向临时文件写入数据完毕，临时文件大小：%d", FileHelper(_tempfile).size());
            //3.删除源文件
            if (FileHelper::removeFile(_datafile) == false)
            {
                DLOG("删除源文件失败！");
                return result;
            }

            //4.修改临时文件名，为源文件名称
            if (FileHelper(_tempfile).renameFile(_datafile) == false)
            {
                DLOG("修改文件名失败！");
                return result;
            }

            //5.返回新的有效数据
            return result;
        }

        std::list<MessagePtr> load()
        {
            std::list<MessagePtr> result;
            //1.加载出文件中所有的有效数据
            FileHelper helper(_datafile);
            size_t offset = 0, msg_size = 0;
            size_t fsize = helper.size();
            //DLOG("准备开始持久化数据，当前文件大小：%d", FileHelper(_datafile).size());
            while (offset < fsize)
            {
                if (helper.read((char*)&msg_size, offset, sizeof(msg_size)) == false)
                {
                    DLOG("读取消息长度失败！！");
                    return {};
                }
                offset += sizeof(msg_size);
                std::string body(msg_size, '\0');
                if (helper.read(&body[0], offset, msg_size) == false)
                {
                    DLOG("读取消息长度失败！！");
                    return {};
                }
                offset += msg_size;

                //将body反序列化成对象
                auto nmp = std::make_shared<Message>();
                nmp->mutable_payload()->ParseFromString(body);
                //如果该消息无效，则不添加，直接处理下一个
                if (nmp->payload().valid() == "0")
                    continue;
                //将有效数据保存至list中
                result.push_back(nmp);
            }
            return result;
        }

    private:
        //3）队列消息的新增持久化
        bool insert(const std::string& filename, MessagePtr& msg)
        {
            //新增数据都是添加在文件末尾
            //1.消息的序列化，得到格式化后的消息
            std::string body = msg->payload().SerializeAsString(); //只需要序列化有效载荷部分
            //2.获取文件长度
            FileHelper helper(filename);
            size_t fsize = helper.size();
            size_t msize = body.size();
            //3.先向文件中写入数据长度
            bool ret = helper.write((char*)&msize, fsize, sizeof(msize));
            fsize += sizeof(msize);
            //4.将数据写入文件指定位置
            ret = helper.write(body.c_str(), fsize, body.size());
            if (ret == false)
            {
                DLOG("向队列数据文件写入数据失败！");
                return false;
            }
            //5.更新msg中的实际存储信息
            msg->set_offset(fsize);
            msg->set_length(body.size());
            return true;
        }

    private:    
        std::string _qname;     // 1）队列名
        std::string _datafile;  // 2）队列消息的存储文件名
        std::string _tempfile;  // 3）队列消息的临时交换文件名
    };

    //队列消息管理
    class QueueMessage
    {
    public:
        using ptr = std::shared_ptr<QueueMessage>;

        QueueMessage(std::string& basedir, const std::string& qname)
            : _mapper(basedir, qname)
            , _qname(qname)
            , _valid_count(0)
            , _total_count(0)
        {
        }

        void recovery()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _msgs = _mapper.gc();

            for (auto& msg : _msgs)
            {
                _durable_msgs.insert({msg->payload().properties().id(), msg});
            }
            _valid_count = _total_count = _durable_msgs.size();
        }

        // 1）新增消息
        bool insert(const BasicProperties* bp, const std::string& body, bool queue_is_durable)
        {
            //1.构造对象
            MessagePtr msg = std::make_shared<Message>();
            msg->mutable_payload()->set_body(body);
            if (bp != nullptr)
            {
                DeliveryMode mode = queue_is_durable ? bp->delivery_mode() : DeliveryMode::UNDURABLE;
                msg->mutable_payload()->mutable_properties()->set_id(bp->id());
                msg->mutable_payload()->mutable_properties()->set_delivery_mode(bp->delivery_mode());
                msg->mutable_payload()->mutable_properties()->set_routing_key(bp->routing_key());
            }
            else  
            {
                DeliveryMode mode = queue_is_durable ? DeliveryMode::DURABLE : DeliveryMode::UNDURABLE;
                msg->mutable_payload()->mutable_properties()->set_id(UUIDHelper::uuid());
                msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                msg->mutable_payload()->mutable_properties()->set_routing_key("");
            }

            std::unique_lock<std::mutex> lock(_mutex);
            //2.判断消息是否需要持久化存储
            if (msg->payload().properties().delivery_mode() == DeliveryMode::DURABLE)
            {
                msg->mutable_payload()->set_valid("1");
                //3.进行持久化存储
                if (_mapper.insert(msg) == false)
                {
                    DLOG("持久化消息 %s 失败", body.c_str());
                    return false;
                }
                _valid_count += 1;
                _total_count += 1;
                _durable_msgs.insert({msg->payload().properties().id(), msg});
            }
            //4.内存的管理
            _msgs.push_back(msg);
            return true;
        }

        // 2）移除指定消息(用于收到确认消息后对消息删除)
        bool remove(const std::string& msg_id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            //1.从待确认队列中查找消息
            auto it = _waitack_msgs.find(msg_id);
            if (it == _waitack_msgs.end())
            {
                DLOG("没有找到要删除的消息");
                return true;
            }
            //2.根据消息的持久化模式，决定是否需要删除持久化信息
            if (it->second->payload().properties().delivery_mode() == DeliveryMode::DURABLE)
            {
                //3.删除持久化信息
                _mapper.remove(it->second);
                _durable_msgs.erase(msg_id);
                _valid_count -= 1;

                //4.判断是否需要垃圾回收
                if (_total_count >= 2000 && _valid_count * 1.0 / _total_count <= 0.5)
                {
                    gc();
                }
            }
            //5.删除内存中的信息
            _waitack_msgs.erase(msg_id);
            return true;
        }

        // 3）获取队首消息（获取的同时将消息加入待确认队列）
        MessagePtr front()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            //从带推送消息链表中获取一条队首消息
            if (_msgs.empty())
            {
                //DLOG("待推送消息队列中没有消息!");
                return {};
            }

            MessagePtr msg = _msgs.front();
            _msgs.pop_front();
            if (msg.get() == nullptr)
                return {};
            //将该消息添加至待确认hash表中，等到消息得到确认后再删除。
            _waitack_msgs.insert({msg->payload().properties().id(), msg});
            return msg;
        }
    
        // 4）获取队列待消费消息数量
        size_t getable_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _msgs.size();
        }

        // 5）获取队列待确认消息数量
        size_t waitack_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _waitack_msgs.size();
        }

        // 6）获取持久化消息数量
        size_t durable_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _durable_msgs.size();
        }

        // 7）总的消息数量
        size_t total_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _total_count;  
        }

        // 8）销毁队列所有消息
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _mapper.removeMsgFile();
            _msgs.clear();
            _durable_msgs.clear();
            _waitack_msgs.clear();
            _valid_count = 0;
            _total_count = 0;
        }

    private:
        void gc()
        {
            //1.进行垃圾回收，获取垃圾回收后，有效消息链表
            std::list<MessagePtr> msgs = _mapper.gc();
            for (auto& msg : msgs)
            {
                auto it = _durable_msgs.find(msg->payload().properties().id());
                if (it == _durable_msgs.end())
                {
                    DLOG("垃圾回收后，有一条持久化消息，在内存中没有进行管理!");
                    _durable_msgs.insert({msg->payload().properties().id(), msg});
                    _msgs.push_back(msg); //重新添加到推送链表
                }

                //2.更新每一条消息的实际存储位置
                it->second->set_offset(msg->offset());
                it->second->set_length(msg->length());
            }
            //3.更新当前的有效消息数量和总的持久化消息数量
            _valid_count = _total_count = msgs.size();
        }

    private:
        std::mutex _mutex;
        std::string _qname;   // 1）队列名称
        size_t _valid_count;  // 2）有效消息数量
        size_t _total_count;  // 3）已经持久化消息总量
        MessageMapper _mapper;    // 4）持久化管理句柄
        std::list<MessagePtr> _msgs;   // 5）待推送消息链表（头插尾删模拟队列）
        std::unordered_map<std::string, MessagePtr> _durable_msgs;  // 6）持久化消息 hash
        std::unordered_map<std::string, MessagePtr> _waitack_msgs;  // 7）待确认消息 hash
    };

    class MessageManager
    {
    public:
        using ptr = std::shared_ptr<MessageManager>;
        MessageManager(const std::string& basedir) //目录
            : _basedir(basedir)
        {
            //该类在重启时应该将队列管理信息数据从磁盘中读取到内存。
            //但是需要所有的队列名称
            //而队列名称获取可以通过mq_queue.hpp中MsgQueueManager的getAll获取
            //但是这样做会将两个模块耦合在一起，所以在这里不用恢复，而是将该工作放到虚拟机部分完成
        }

        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            for (auto& qmsg : _queue_msgs)
                qmsg.second->clear();
        }

        //1）初始化新建队列的消息管理句柄，并创建消息存储文件
        void initQueueMessage(const std::string& qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it != _queue_msgs.end())
                    return;
                qmp = std::make_shared<QueueMessage>(_basedir, qname);
                _queue_msgs.insert({qname, qmp});
            }
            qmp->recovery();
        }

        // 2）删除队列的消息管理句柄，以及消息存储文件
        void destroyQueueMessage(const std::string& qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                    return;
                qmp = it->second;
                _queue_msgs.erase(qname);
            }
            qmp->clear();
        }

        // 3）向指定队列新增消息
        bool insert(const std::string& qname, const BasicProperties* bp, const std::string& body, bool queue_is_durable)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("向队列%s新增消息失败：没有找到消息句柄!", qname.c_str());
                    return false;
                }
                qmp = it->second;
            }
            qmp->insert(bp, body, queue_is_durable);
            return true;
        }
        
        // 4）获取指定队列队首消息
        MessagePtr front(const std::string& qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("获取队列%s首消息失败：没有找到消息句柄!", qname.c_str());
                    return {};
                }
                qmp = it->second;
            }
            return qmp->front();
        }

        // 5）确认指定队列待确认消息
        void ack(const std::string& qname, const std::string msg_id)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("确认队列%s中%s消息失败：没有找到消息句柄!", qname.c_str(), msg_id.c_str());
                    return;
                }
                qmp = it->second;
            }
            qmp->remove(msg_id);
            return;
        }

        // 4）获取队列待消费消息数量
        size_t getable_count(const std::string& qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("获取队列%s中待推送消息失败：没有找到消息句柄!", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->getable_count();
        }

        // 5）获取队列待确认消息数量
        size_t waitack_count(const std::string& qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("获取队列%s中待确认消息失败：没有找到消息句柄!", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->waitack_count();
        }
        // 6）获取持久化消息数量
        size_t durable_count(const std::string& qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("获取队列%s中持久化u消息失败：没有找到消息句柄!", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->durable_count();
        }
        // 7）总的消息数量
        size_t total_count(const std::string& qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("获取队列%s中总消息数量失败：没有找到消息句柄!", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->total_count();
        }

    private:
        std::mutex _mutex;
        std::string _basedir;
        std::unordered_map<std::string, QueueMessage::ptr> _queue_msgs;
    };
}